
package mvcc

import (
	"encoding/binary"
	"time"

	"go.etcd.io/etcd/server/v3/storage/schema"
	"go.uber.org/zap"
)

// store.scheduleCompaction（）方法中完成了对BoltDB中储存的键值对的压缩，
// 在该方法会通过UnsafeRange（）方法批量查询待删除的Key(revision），
// 然后逐个调用UnsafeDelete（）方法进行删除，全部的待压缩Key被处理完成之后，
// 会向meteBucket中写入此次压缩的相关信息，其中Key为“finishedCompactRev”。
func (s *store) scheduleCompaction(compactMainRev int64, keep map[revision]struct{}) bool {
	totalStart := time.Now()
	defer func() { dbCompactionTotalMs.Observe(float64(time.Since(totalStart) / time.Millisecond)) }()
	keyCompactions := 0
	defer func() { dbCompactionKeysCounter.Add(float64(keyCompactions)) }()
	defer func() { dbCompactionLast.Set(float64(time.Now().Unix())) }()
	// 这里将compactMainRev+l写入end([]byte类型）中（Compact（）方法中已经更新过compactMainRev),
	// 作为范围查询的结束key(revision)
	end := make([]byte, 8)
	binary.BigEndian.PutUint64(end, uint64(compactMainRev+1))

	batchNum := s.cfg.CompactionBatchLimit	// 范围查询的上限
	batchInterval := s.cfg.CompactionSleepInterval
	// 范围查询的起始key(revision)
	last := make([]byte, 8+1+8)
	for {
		var rev revision

		start := time.Now()

		tx := s.b.BatchTx()	// 获取读写事务
		tx.Lock()	// 加锁
		// 调用UnsafeRange（）方法对keyBucket进行范围查询
		keys, _ := tx.UnsafeRange(schema.Key, last, end, int64(batchNum))
		// 遍历上述查询到的Key(revision），并逐个调用UnsafeDelete（）方法进行删除
		for _, key := range keys {
			rev = bytesToRev(key)
			if _, ok := keep[rev]; !ok {
				tx.UnsafeDelete(schema.Key, key)
				keyCompactions++
			}
		}

		if len(keys) < batchNum {	// 最后一次批量处理
			// 将compactMainRev封装成revision，并写入metaBucket中，其Key为“finishedCompactRev”
			UnsafeSetFinishedCompact(tx, compactMainRev)
			// 读写事务使用完毕，解锁
			tx.Unlock()
			s.lg.Info(
				"finished scheduled compaction",
				zap.Int64("compact-revision", compactMainRev),
				zap.Duration("took", time.Since(totalStart)),
			)
			return true
		}

		tx.Unlock()
		// 更新last，下次范围查询的起始key依然是last
		revToBytes(revision{main: rev.main, sub: rev.sub + 1}, last)
		// 立即提交压缩删除，而不是让它们累积在写入缓冲区中
		s.b.ForceCommit()
		dbCompactionPauseMs.Observe(float64(time.Since(start) / time.Millisecond))

		select {	// 每次范围处理结束之后
		case <-time.After(batchInterval):
		case <-s.stopc:
			return false
		}
	}
}
